/*
 * Copyright 2011 Rolf Aden
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License. 
 */

package com.ec;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Query.FilterOperator;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.tools.mapreduce.contrib.DatastoreMapper;

/**
 * Mapper to copy all entities of one kind to an other (rename).
 * 
 * @author Rolf Aden
 */
public class RenameEntityKindMapper extends DatastoreMapper {
	
	private static final Logger LOG = Logger.getLogger(RenamePropertyMapper.class.getName());
	
	static final String PHASE = "ec.phase";
	static final String NEW_KIND = "ec.newKind";
	static final String USE_TRANSACTIONS = "ec.useTransactions";
	
	private DatastoreService ds = DatastoreServiceFactory.getDatastoreService();
	private String newKind;
	private boolean useTransactions = true;
	
	@Override
	public void taskSetup(Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException ,InterruptedException {
		super.taskSetup(context);

		//	TODO input validation
		this.newKind = context.getConfiguration().get(NEW_KIND);
		this.useTransactions = Boolean.parseBoolean(context.getConfiguration().get(USE_TRANSACTIONS));
	}

	@Override
	public void map(Key key, Entity entity, Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException, InterruptedException 
	{
		if (LOG.isLoggable(Level.FINER)) {
			LOG.finer("Processing key: " + key);
		}
		
		Key newKey = entity.getKey().getName() != null
    		? KeyFactory.createKey(this.newKind, entity.getKey().getName())
    		: KeyFactory.createKey(this.newKind, entity.getKey().getId());
    		
		//	if we do not use transactions, an already elsewhere created newEntity 
		//	will be overridden
		Transaction tx = this.useTransactions
			? ds.beginTransaction()
			: null;
		try {
			//	Do we have already a new entity?
			//	Since we do not update an existing entity key only query
			//	is sufficent
			Query query = new Query(this.newKind)
				.addFilter(Entity.KEY_RESERVED_PROPERTY, FilterOperator.EQUAL, newKey)
				.setKeysOnly();

			if (ds.prepare(tx, query).asSingleEntity() == null) {
				//	We do not have a new entity, 
				//	create a new one and persist it
				Entity newEntity = new Entity(newKey); 
				newEntity.setPropertiesFrom(entity);
				ds.put(tx, newEntity);
			}
			
			if (tx != null) {
				tx.commit();
			}

			// we can assume here, the new entity has been created
			ds.delete(key);
			
		} finally {
			if (tx != null && tx.isActive()) {
				tx.rollback();
			}
		}
	}
}
